package com.atguigu.gmall.realtime.utils;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.Semantic;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.util.Properties;

/**
 * 操作 kafka 的工具类
 *
 * @author lvbingbing
 * @date 2022-03-12 22:17
 */
public class KafkaUtils {

    private static final String KAFKA_SERVER = "hadoop102:9092,hadoop103:9092,hadoop104:9092";

    private static final String DEFAULT_TOPIC = "default_topic";

    private KafkaUtils() {
    }

    /**
     * 获取 kafka 消费者
     *
     * @param topic   主题
     * @param groupId 消费者组id
     * @return <br>
     */
    public static FlinkKafkaConsumer<String> getKafkaSource(String topic, String groupId) {
        Properties prop = new Properties();
        prop.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        prop.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_SERVER);
        return new FlinkKafkaConsumer<>(topic, new SimpleStringSchema(), prop);
    }

    /**
     * 获取 kafka 生产者
     * <p>
     * 只能保证至少一次，无法保证精准一次性
     * <p>
     * todo 问题记录：BaseLogApp.java 中 第 7 步：将不同流的数据写 kafka 的 dwd 不同的主题中
     * <p>
     * 改成使用精准一次性时会报以下错误：
     * javax.management.InstanceAlreadyExistsException: kafka.producer:type=app-info,id="producer-Map -> Process -> (Sink: Print to Std. Out, Sink: Print to Std. Out, Sink: Print to Std. Out, Sink: Unnamed, Sink: Unnamed, Sink: Unnamed)-a830ee5c0bf04af591ad3d80da86825c-18"
     * <p>
     * 暂时未查询到原因，改为调用该方法时(保证至少一次)，就不会报错了
     * <p>
     * todo 可能与检查点配置有关系，注释掉检查点相关代码就不报错了。
     * <p>
     * todo 该方法暂时不废弃，待确定原因之后再决定是否废弃
     *
     * @param topic 主题
     * @return kafka 生产者
     * @see FlinkKafkaProducer#FlinkKafkaProducer(java.lang.String, org.apache.flink.api.common.serialization.SerializationSchema, java.util.Properties, java.util.Optional)
     * @see Semantic
     * @see KafkaUtils#getKafkaSinkExactlyOnce(java.lang.String)
     */
    public static FlinkKafkaProducer<String> getKafkaSinkAtLeastOnce(String topic) {
        return new FlinkKafkaProducer<>(KAFKA_SERVER, topic, new SimpleStringSchema());
    }

    /**
     * 将 String 序列化为 ProducerRecords，获取 kafka 生产者
     * <p>
     * 保证精准一次性
     *
     * @param topic 主题
     * @return kafka 生产者
     */
    public static FlinkKafkaProducer<String> getKafkaSinkExactlyOnce(String topic) {
        KafkaSerializationSchema<String> stringSerializationSchema = new KafkaSerializationSchema<String>() {
            @Override
            public ProducerRecord<byte[], byte[]> serialize(String element, @Nullable Long timestamp) {
                return new ProducerRecord<>(topic, element.getBytes());
            }
        };
        return getKafkaSinkExactlyOnce(stringSerializationSchema);

    }

    /**
     * 将 T 类型的值序列化为 ProducerRecords，获取 kafka 生产者
     *
     * @param kafkaSerializationSchema 可序列化的序列化模式，用于将用户对象转换为支持键/值消息的 kafka 可消费字节 []
     * @param <T>                      被序列化的值的类型
     * @return kafka 生产者
     */
    public static <T> FlinkKafkaProducer<T> getKafkaSinkExactlyOnce(KafkaSerializationSchema<T> kafkaSerializationSchema) {
        Properties props = new Properties();
        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_SERVER);
        props.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 1000 * 60 * 15 + "");
        return new FlinkKafkaProducer<>(DEFAULT_TOPIC, kafkaSerializationSchema, props, Semantic.EXACTLY_ONCE);
    }

    /**
     * 获取 kafka 相关连接信息到 ddl
     *
     * @param topic   <br>
     * @param groupId <br>
     * @return
     */
    public static String getConnectInfo(String topic, String groupId) {
        return "  'connector' = 'kafka', " +
                " 'topic' = '" + topic + "', " +
                " 'properties.bootstrap.servers' = '" + KAFKA_SERVER + "', " +
                " 'properties.group.id' = '" + groupId + "', " +
                " 'scan.startup.mode' = 'latest-offset', " +
                " 'format' = 'json' ";
    }
}